package com.rs.utils;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONObject;
import com.rs.config.LoggableStatement;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.sql.rowset.serial.SerialBlob;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.sql.*;
import java.util.*;
import java.util.Date;

@Slf4j
@Component
public class JdbcTemplateUtil {

    @Resource
    private DruidDataSource druidDataSource;

    /**
     * 保存数据到达梦
     * @param tableName 表名
     * @param columnMap 字段名-字段值映射
     * @param columnTypeMap 字段名-字段flink类型
     */
    public boolean saveDataToDm(String tableName, Map<String, Object> columnMap,
                             Map<String, String> columnTypeMap) {
        // 字段集合
        Set<String> columnSet = columnMap.keySet();
        String sqlTemplate = buildInsertSqlTemplate(tableName, columnSet);
        if (StrUtil.isEmpty(sqlTemplate)) {
            log.error("表名[{}]或者表字段为空", tableName);
            return false;
        }

        // 获取数据库连接
        Connection conn = null;
        LoggableStatement ps = null;
        try {
            conn = druidDataSource.getConnection();
            // 不允许自动提交,这是我们首先要做的
            conn.setAutoCommit(false);
            ps = new LoggableStatement(conn, sqlTemplate);

            // 填充值
            int idx = 1;
            for (String column : columnSet) {
                String columnType = columnTypeMap.get(column);
                if (columnType == null) {
                    idx++;
                    continue;
                }
                columnType = columnType.toUpperCase();
                switch (columnType) {
                    case "SHORT":
                    case "SMALLINT":
                    case "INT":
                        ps.setInt(idx, MapUtil.getInt(columnMap, column));
                        break;
                    case "LONG":
                    case "BIGINT":
                        ps.setLong(idx, MapUtil.getLong(columnMap, column));
                        break;
                    case "BOOLEAN":
                        ps.setBoolean(idx, MapUtil.getBool(columnMap, column));
                        break;
                    case "FLOAT":
                        ps.setFloat(idx, MapUtil.getFloat(columnMap, column));
                        break;
                    case "DOUBLE":
                        ps.setDouble(idx, MapUtil.getDouble(columnMap, column));
                        break;
                    case "DECIMAL":
                    case "DECIMAL(38,18)":
                        ps.setBigDecimal(idx, new BigDecimal(MapUtil.getStr(columnMap, column)));
                        break;
                    case "VARCHAR":
                    case "CHAR":
                    case "STRING":
                        ps.setString(idx, MapUtil.getStr(columnMap, column));
                        break;
                    case "DATE":
                    case "TIME":
                    case "TIMESTAMP":
                        Date date = MapUtil.getDate(columnMap, column);
                        if (date != null) {
                            ps.setString(idx, DateUtil.format(date, "yyyy-MM-dd HH:mm:ss"));
                        }
                        break;
                    case "BYTES":
                        Object blobObj = columnMap.get(column);
                        if (blobObj != null) {
                            // 设置Blob
                            Blob b = new SerialBlob(JSONObject.toJSONString(blobObj).getBytes(StandardCharsets.UTF_8));
                            ps.setBlob(idx, b);
                        }
                        break;
                    default:
                        ps.setObject(idx, columnMap.get(column));
                }
                idx++;
            }
            ps.execute();
            log.info("==> Executing SQL: " + ps.getQueryString() );
            conn.commit();
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        } finally {
            release(conn, ps);
        }
        return true;
    }

    /**
     * 根据主键删除数据
     * @param tableName 表名
     * @param primaryKey 主键名称
     * @param primaryKeyVal 主键值
     */
    public void deleteDataByPrimaryKey(String tableName, String primaryKey, String primaryKeyVal) {
        String sqlTemplate = buildDeleteSqlTemplate(tableName, Collections.singletonList(primaryKey), Collections.singletonList(primaryKeyVal));

        // 获取数据库连接
        Connection conn = null;
        LoggableStatement ps = null;

        try {
            conn = druidDataSource.getConnection();
            // 不允许自动提交,这是我们首先要做的
            conn.setAutoCommit(false);
            ps = new LoggableStatement(conn, sqlTemplate);

            ps.execute();
            log.info("==> Executing SQL: " + ps.getQueryString() );
            conn.commit();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            release(conn, ps);
        }
    }

    /**
     * 构建根据k-v删除SQL模板 k-v顺序需要保持一致
     * @param tableName 表名
     * @param keyList 键
     * @param valList 值
     * @return String
     */
    private static String buildDeleteSqlTemplate(String tableName, List<String> keyList, List<String> valList) {
        // 根据表名字段构建执行SQL
        StringBuilder sql = new StringBuilder();

        sql.append("DELETE FROM ").append(tableName).append(" WHERE ");

        for (int i = 0; i < keyList.size(); i++) {
            sql.append(keyList.get(i)).append(" = ").append("'").append(valList.get(i)).append("'");
            sql.append(" AND ");
        }

        sql.setLength(sql.length() - 5);

        return sql.toString();
    }

    /**
     * 构建插入SQL模板
     * @param tableName 表名
     * @param columnList 字段名集合
     * @return String
     */
    private String buildInsertSqlTemplate(String tableName, Set<String> columnList) {
        if (StrUtil.isEmpty(tableName) || CollectionUtil.isEmpty(columnList)) {
            return null;
        }

        // 根据表名字段构建执行SQL
        StringBuilder sql = new StringBuilder();
        sql.append("INSERT INTO \"").append(tableName).append("\" (");
        // 拼接字段
        for (String column : columnList) {
            sql.append(" \"").append(column).append("\",");
        }
        sql.setLength(sql.length() - 1);
        sql.append(" ) VALUES (");
        // 拼接值
        for (String ignored : columnList) {
            sql.append(" ?,");
        }
        sql.setLength(sql.length() - 1);
        sql.append(" )");

        return sql.toString();
    }

    /**
     * 释放资源
     * @param conn conn
     * @param ps ps
     */
    private void release(Connection conn, PreparedStatement ps) {
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (ps != null) {
            try {
                ps.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }
}
